import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

import java.util

object HelloSparkStream {

  // 设置 Hadoop 相关环境变量
  System.setProperty("hadoop.home.dir", "F:\\Hadoop\\hadoop-3.3.6-windows-bin")
  System.setProperty("HADOOP_USER_NAME", "root")

  def main(args: Array[String]): Unit = {

    // 创建 sparkconf 配置项 本地，应用程序名字
    val conf = new SparkConf().setMaster("local[*]").setAppName("helloStream")
    // 创建 sparkStreaming  加载conf （ 实时的数据流）  ssc
    val ssc = new StreamingContext(conf, Seconds(2))

    // 设置日志级别为 error
    ssc.sparkContext.setLogLevel("error")

    // 设置检查点目录
    ssc.checkpoint("F:\\SparkCheckpoint")

    // kafka 配置
    val kfkParams = Map[String, Object](
      "bootstrap.servers" -> "123.56.187.176:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "niit",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    // topic
    val topicName = Array("stuInfo")

    // kafkaUtils  导入一个依赖 spark-streaming-kafka
    val streamRdd = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent, // 位置策略
      Subscribe[String, String](topicName, kfkParams)  // 订阅topic的名字
    )

    // 对 Kafka 流数据进行处理
    val lines = streamRdd.map(_.value())

    // 窗口函数
    val windowDuration = Seconds(10)  // 窗口持续时间
    val slideDuration = Seconds(2)    // 窗口滑动间隔

    val wordCounts = lines.flatMap(_.split("\t"))
      .map(word => (word, 1))
      .reduceByKeyAndWindow(_ + _, _ - _, windowDuration, slideDuration)

    // 输出结果
    wordCounts.foreachRDD { rdd =>
      println("==========数据结果===========")
      rdd.foreach(println)
    }

    // producer 配置项
    val property = new util.HashMap[String, Object]()
    property.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.192.138:9092")
    property.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    property.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")

    // 处理 Kafka 数据
    wordCounts.foreachRDD { rdd =>
      if (!rdd.isEmpty()) {
        // 将结果发送回 Kafka
        rdd.foreach { obj =>
          val kfkProducer = new KafkaProducer[String, String](property)
          kfkProducer.send(new ProducerRecord[String, String]("xiao77", obj.toString()))
          kfkProducer.close()
        }
      }
    }

    ssc.start()  // 开启 ssc
    ssc.awaitTermination() // 等待数据输入
  }
}